import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;
import org.apache.flink.types.Row;

import java.util.Arrays;

import static org.apache.flink.table.api.Expressions.$;

public class OuterJoin {



    public static void main(String[] args) throws Exception
    {


        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        BatchTableEnvironment tableEnv = BatchTableEnvironment.create(env);



        DataSet<Order> ds1 = env.fromCollection(Arrays.asList(
                new Order(1L, "beer", 3),
                new Order(3L, "rubber", 2),
                new Order(1L, "diaper", 4)
        ));


        DataSet<Order_copy> ds2 = env.fromCollection(Arrays.asList(
                new Order_copy(2L, "apple", 36),
                new Order_copy(3L, "orange", 2),
                new Order_copy(2L, "pear", 34)
        ));


        Table left =  tableEnv.fromDataSet(ds1, "user, product, amount");
        Table right = tableEnv.fromDataSet(ds2, "userid, fruit, number");

        Table leftOuterResult = left.leftOuterJoin(right, $("user").isEqual($("userid")))
            .select($("user"), $("product"), $("fruit"));

        Table rightOuterResult = left.rightOuterJoin(right, $("user").isEqual($("userid")))
                .select($("user"), $("product"), $("fruit"));


        Table fullOuterResult = left.fullOuterJoin(right, $("user").isEqual($("userid")))
                .select($("user"), $("product"), $("fruit"));




        tableEnv.toDataSet(fullOuterResult, Row.class).print();


    }

}

